package com.wuwangfu.func;

import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashSet;

/**
 * @Author: jcshen
 * @Date: 2023-06-18
 * @PackageName: com.wuwangfu.func
 * @ClassName: BroadcastProcessFunc
 * @Description:
 * @Version: 1.0.0
 */
public class BroadcastProcessFunc extends KeyedBroadcastProcessFunction<
                Tuple2<String,String>, //key
                Tuple3<String, String, String>,//input--->data
                Tuple3<String, String, String>,//input--->broadcast
                Tuple4<String, String, Long,Long>//output
                > {

    //定义状态，不能transient，需要传入进来，会参与序列化和反序列化，否则会获取不到
    private  MapStateDescriptor<String, String> dicStateDesctor;

    private transient ValueState<Long> actCountState;
    private transient ValueState<HashSet<String>> userCountState;

    //通过 构造方法 或 set方法传递，使用构造方法传递更佳方便；添加有参和无参构造方法，不然会有反射问题
    public BroadcastProcessFunc(){}
    public BroadcastProcessFunc(MapStateDescriptor<String, String> dicStateDesctor) {
        this.dicStateDesctor = dicStateDesctor;
    }


    @Override
    public void open(Configuration parameters) throws Exception {
        ValueStateDescriptor<Long> actCountDesctor = new ValueStateDescriptor<>("act-count-state", Long.class);
        actCountState = getRuntimeContext().getState(actCountDesctor);

        ValueStateDescriptor<HashSet<String>> userStateDesctor = new ValueStateDescriptor<>("user-dist-state",
                TypeInformation.of(new TypeHint<HashSet<String>>() {}));
        userCountState = getRuntimeContext().getState(userStateDesctor);
    }


    /**
     * 处理输入的每一条业务数据
     *
     * @param value
     * @param ctx
     * @param out
     * @throws Exception
     */
    @Override
    public void processElement(Tuple3<String, String, String> value, ReadOnlyContext ctx, Collector<Tuple4<String, String, Long,Long>> out) throws Exception {

        //获取广播状态
        ReadOnlyBroadcastState<String, String> broadcastState = ctx.getBroadcastState(dicStateDesctor);

        //计算次数
        Long histCount = actCountState.value();
        if (histCount == null){
            histCount = 0L;
        }
        //累计状态
        long totalCount = histCount + 1;
        //更新状态
        actCountState.update(totalCount);

        //计算人数
        HashSet<String> disUserSet = userCountState.value();
        if (disUserSet == null){
            disUserSet = new HashSet<>();
        }
        //累计状态
        disUserSet.add(value.f0);
        //更新状态
        userCountState.update(disUserSet);
        //输出结果
        String evnetId = value.f2;
        String eventName = broadcastState.get(evnetId);

        out.collect(Tuple4.of(value.f1,eventName,totalCount,(long)disUserSet.size()));
    }


    /**
     * 处理输入的每一条维度数据
     *
     * @param value
     * @param ctx
     * @param out
     * @throws Exception
     */
    @Override
    public void processBroadcastElement(Tuple3<String, String, String> value, Context ctx, Collector<Tuple4<String, String, Long,Long>> out) throws Exception {

        String type = value.f0;
        String eventId = value.f1;
        String eventName = value.f2;
        //在main方法中已经定义了广播状态，使用构造器的方式，传递广播状态，通过状态描述器，获取广播状态
        BroadcastState<String, String> broadcastState = ctx.getBroadcastState(dicStateDesctor);
        //将广播的状态存起来或更新、删除
        if ("DELETE".equals(type)){
            broadcastState.remove(type);
        }else{
            broadcastState.put(eventId,eventName);
        }
    }
}
